package com.zbmy.plugin.im.server.core.zookeeper.server;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.zbmy.plugin.im.server.core.NetShareInstance;

@Component
public class ZKServiceServerRegistry {
	 	private CountDownLatch latch = new CountDownLatch(1);
	    @Value("${registryAddress}")
		private String registryAddress;
	    //注册到zk中，其中data为服务端的 ip:port
	    public void register(String data) {
	        if (data != null) {
	            ZooKeeper zk = connectServer();		// 连接zookeeper服务器
	            if (zk != null) {	// 连接成功，创建节点信息
	                createNode(zk, data);
	            }
	        }
	    }
	    /**
	     * 
	     * 连接zookeeper服务器   
	     * @author 杨乐平    
	     * @Title: connectServer    
	     * @return 
	     * @Return: ZooKeeper 返回值
	     */
	    private ZooKeeper connectServer() {
	        ZooKeeper zk = null;
	        try {
	        	// 连接zookeeper服务器
	           zk = new ZooKeeper(registryAddress, /*Constant.ZK_SESSION_TIMEOUT*/2000, new Watcher() {
	                @Override
	                public void process(WatchedEvent event) {
	                    if (event.getState() == Event.KeeperState.SyncConnected) {
	                    	// 连接成功
	                        latch.countDown();
	                    }
	                }
	            });
	            latch.await();
	        } catch (Exception e) {
	            e.printStackTrace();
	        }
	        return zk;
	    }
	    private void createNode(ZooKeeper zk, String data) {
	        try {
	            byte[] bytes = data.getBytes();
	            zk.create("/zk-node", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	            String path = zk.create("/zk-node/" + NetShareInstance.serverIp, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
	            System.out.println("create zookeeper node path:"+path+" data:"+data);
	        } catch (Exception e) {
	        	e.printStackTrace();
	        }
	    }
	    
	    public static void startUp(){
	    		//int port = 9988;
	    	  //register service in zookeeper 
	    	  ZKServiceServerRegistry zsr = new ZKServiceServerRegistry();
	    	  String serverIp = NetShareInstance.serverIp + ":" + NetShareInstance.frontenServerPort;
	    	  zsr.register(serverIp);
	    }
}
